package com.nami.ucenter.infra.core.middleware.mq;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.ucenter.domain.port.publisher.DomainEventPublisher;
import com.nami.ucenter.domain.port.publisher.request.DownModeConfig;
import com.nami.ucenter.domain.port.publisher.request.DownMsgBatchReq;
import com.nami.ucenter.domain.port.publisher.request.DownMsgRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import static com.nami.gate.external.basic.constants.CommonConst.COLON;


@Slf4j
@RequiredArgsConstructor
@Service
public class DownMsgPublishImpl implements DomainEventPublisher {

    private final RocketMQTemplate rocketMQTemplate;

    @Override
    public boolean publish(String topic, String tag, byte[] msg) {
        TimeInterval timer = DateUtil.timer();
//        log.info("[producer] 准备发送消息. topic={}|{}", topic, tag);

        Message<byte[]> result = MessageBuilder
            .withPayload(msg)
//            .setHeader(RocketMQHeaders.TAGS, tag)
            .build();
        rocketMQTemplate.syncSend(topic + COLON + tag, result);
        log.info("[producer] topic={}, tags={}", topic, tag);
//        rocketMQTemplate.syncSend(topic + COLON + tag, MessageBuilder.withPayload(msg).build());
        if (timer.interval() > 100) {
            log.warn("[producer] mq发送时间超时了。topic={}|{}", topic, tag);
        }
        return Boolean.TRUE;
    }

    @Override
    public boolean publish(String topic, String tag, String msg) {
        TimeInterval timer = DateUtil.timer();
//        log.info("[producer] 准备发送消息. topic={}|{}", topic, tag);

        Message<String> result = MessageBuilder.withPayload(msg)
//            .setHeader(RocketMQHeaders.TAGS, tag)
            .build();
        rocketMQTemplate.syncSend(topic + COLON + tag, result);
        log.info("[producer] topic={}, tags={}", topic, tag);
//        rocketMQTemplate.syncSend(topic + COLON + tag, MessageBuilder.withPayload(msg).build());
        if (timer.interval() > 100) {
            log.warn("[producer] mq发送时间超时了。topic={}|{}", topic, tag);
        }
        return Boolean.TRUE;
    }

    /**
     * flshmore 是包涵flushOne topic 。 是否删除当前发送一条的topic, 改为flushMore. 修改为支持单发，多发
     *
     * @param request
     * @param config
     * @return
     * @throws Exception
     */
    @Override
    public boolean flushOne(DownMsgRequest request, DownModeConfig config) throws Exception {
        switch (config.getMode()) {
            case "mq":
                String body = JsonUtils.toJsonString(request);
//                Message message = new Message(DownTopicConst.FLUSH_ONE_TOPIC, config.getMqTags(), body.getBytes("utf-8"));
                Message<String> result = MessageBuilder.withPayload(body).setHeader(RocketMQHeaders.TAGS, config.getMqTags()).build();
                rocketMQTemplate.syncSend(DownTopicConst.FLUSH_ONE_TOPIC + COLON + config.getMqTags(), result);
                log.info("[下行消息] flushOne topic={}, tags={}", DownTopicConst.FLUSH_ONE_TOPIC, config.getMqTags());
//                rocketMQTemplate.syncSend(DownTopicConst.FLUSH_ONE_TOPIC+COLON + config.getMqTags(), MessageBuilder.withPayload(JsonUtils.toJsonString(request)).build());
                break;
            case "dubbo":
//                ForwardPacketUtil.flushPacket(request, config.getDubboIp(), config.getDubboPort());
                break;
            default:
                log.error("[qos] 下行通信方式没找到...");

        }
        return Boolean.TRUE;
    }

    @Override
    public boolean flushMore(DownMsgBatchReq request, DownModeConfig config) throws Exception {
        switch (config.getMode()) {
            case "mq":
                String body = JsonUtils.toJsonString(request);
                Message<String> result = MessageBuilder.withPayload(body).setHeader(RocketMQHeaders.TAGS, config.getMqTags()).build();
                rocketMQTemplate.syncSend(DownTopicConst.FLUSH_MORE_TOPIC + COLON + config.getMqTags(), result);
                log.info("[下行消息] flushMore topic={}, tags={}", DownTopicConst.FLUSH_MORE_TOPIC, config.getMqTags());
                //                rocketMQTemplate.syncSend(DownTopicConst.FLUSH_MORE_TOPIC + COLON + config.getMqTags(), MessageBuilder.withPayload(JsonUtils.toJsonString(req)).build());
                break;
            case "dubbo":
//                ForwardPacketUtil.flushMore(request, config.getDubboIp(), config.getDubboPort());
                break;
            default:
                log.error("[qos] 下行通信方式没找到...");
                return false;

        }
        return Boolean.TRUE;
    }


}
